/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl.statistics.analyzer;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import spring.cloud.tasks.api.domain.AbnormalJob;
import spring.cloud.tasks.api.domain.AbnormalShardingState;
import spring.cloud.tasks.api.domain.NameSpaceConfiguration;
import spring.cloud.tasks.api.domain.TaskType;
import spring.cloud.tasks.api.service.ReportAlarmService;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.helper.DashboardConstants;
import spring.cloud.tasks.api.service.helper.DashboardServiceHelper;
import spring.cloud.tasks.api.service.impl.TaskServiceImpl;
import spring.cloud.tasks.api.utils.CronExpression;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class OutdatedNoRunningJobAnalyzer {

    private static final Logger log = LoggerFactory.getLogger(OutdatedNoRunningJobAnalyzer.class);
    private final List<AbnormalJob> outdatedNoRunningJobs = new ArrayList<>();
    private final List<AbnormalJob> needReportAlarmJobs = new ArrayList<>();
    private Map<String/** domainName_jobName_shardingItemStr **/
            , AbnormalShardingState /** abnormal sharding state */> abnormalShardingStateCache = new ConcurrentHashMap<>();
    private ReportAlarmService reportAlarmService;
    private TaskService taskService;

    private static boolean isCronJob(ZooKeeper zooKeeper, String jobName) {
        String jobType = zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "jobType"));
        return TaskType.isCron(TaskType.getJobType(jobType));
    }

    private static boolean isEnabledPath(ZooKeeper zooKeeper,
                                         AbnormalJob abnormalJob) {
        String enabledPath = TaskNodePath.getConfigNodePath(abnormalJob.getJobName(), "enabled");
        return Boolean.parseBoolean(zooKeeper.getData(enabledPath));
    }


    private static long getLastCompleteTime(ZooKeeper zooKeeper, String jobName,
                                            String shardingItemStr) {
        String lastCompleteTimePath = TaskNodePath.getExecutionNodePath(jobName, shardingItemStr, "lastCompleteTime");
        String data = zooKeeper.getData(lastCompleteTimePath);
        return StringUtils.isBlank(data) ? 0 : Long.parseLong(data.trim());
    }

    /**
     * 该时间是否在作业暂停时间段范围内。
     * <p>
     * 特别的，无论pausePeriodDate，还是pausePeriodTime，如果解析发生异常，则忽略该节点，视为没有配置该日期或时分段。
     *
     * @return 该时间是否在作业暂停时间段范围内。
     */
    private static boolean isInPausePeriod(Date date, String pausePeriodDate, String pausePeriodTime,
                                           TimeZone timeZone) {
        Calendar calendar = Calendar.getInstance(timeZone);
        calendar.setTime(date);
        int iMon = calendar.get(Calendar.MONTH) + 1; // Calendar.MONTH begin from 0.
        int d = calendar.get(Calendar.DAY_OF_MONTH);
        int h = calendar.get(Calendar.HOUR_OF_DAY);
        int m = calendar.get(Calendar.MINUTE);

        boolean pausePeriodDateIsEmpty = (pausePeriodDate == null || pausePeriodDate.trim().isEmpty());
        boolean dateIn = false;
        if (!pausePeriodDateIsEmpty) {
            dateIn = isDateInPausePeriodDate(iMon, d, pausePeriodDate);
        }

        boolean timeIn = false;
        boolean pausePeriodTimeIsEmpty = (pausePeriodTime == null || pausePeriodTime.trim().isEmpty());
        if (!pausePeriodTimeIsEmpty) {
            timeIn = isTimeInPausePeriodTime(h, m, pausePeriodTime);
        }

        if (pausePeriodDateIsEmpty) {
            if (pausePeriodTimeIsEmpty) {
                return false;
            } else {
                return timeIn;
            }
        } else {
            if (pausePeriodTimeIsEmpty) {
                return dateIn;
            } else {
                return dateIn && timeIn;
            }
        }
    }

    private static boolean isDateInPausePeriodDate(int m, int d, String pausePeriodDate) {
        boolean dateIn = false;

        String[] periodsDate = pausePeriodDate.split(",");
        if (periodsDate == null) {
            return dateIn;
        }
        for (String period : periodsDate) {
            String[] tmp = period.trim().split("-");
            if (tmp == null || tmp.length != 2) {
                dateIn = false;
                break;
            }
            String left = tmp[0].trim();
            String right = tmp[1].trim();
            String[] sMdLeft = left.split("/");
            String[] sMdRight = right.split("/");
            if (sMdLeft != null && sMdLeft.length == 2 && sMdRight != null && sMdRight.length == 2) {
                try {
                    int iMLeft = Integer.parseInt(sMdLeft[0]);
                    int dLeft = Integer.parseInt(sMdLeft[1]);
                    int iMRight = Integer.parseInt(sMdRight[0]);
                    int dRight = Integer.parseInt(sMdRight[1]);
                    boolean isBiggerThanLeft = m > iMLeft || (m == iMLeft && d >= dLeft);
                    boolean isSmallerThanRight = m < iMRight || (m == iMRight && d <= dRight);
                    dateIn = isBiggerThanLeft && isSmallerThanRight;
                    if (dateIn) {
                        break;
                    }
                } catch (NumberFormatException e) {
                    dateIn = false;
                    break;
                }
            } else {
                dateIn = false;
                break;
            }
        }
        return dateIn;
    }

    private static boolean isTimeInPausePeriodTime(int hour, int min, String pausePeriodTime) {
        boolean timeIn = false;
        String[] periodsTime = pausePeriodTime.split(",");
        if (periodsTime == null) {
            return timeIn;
        }
        for (String period : periodsTime) {
            String[] tmp = period.trim().split("-");
            if (tmp == null || tmp.length != 2) {
                timeIn = false;
                break;
            }
            String left = tmp[0].trim();
            String right = tmp[1].trim();
            String[] hmLeft = left.split(":");
            String[] hmRight = right.split(":");
            if (hmLeft != null && hmLeft.length == 2 && hmRight != null && hmRight.length == 2) {
                try {
                    int hLeft = Integer.parseInt(hmLeft[0]);
                    int mLeft = Integer.parseInt(hmLeft[1]);
                    int hRight = Integer.parseInt(hmRight[0]);
                    int mRight = Integer.parseInt(hmRight[1]);
                    boolean isBiggerThanLeft = hour > hLeft || (hour == hLeft && min >= mLeft);
                    boolean isSmallerThanRight = hour < hRight || (hour == hRight && min <= mRight);
                    timeIn = isBiggerThanLeft && isSmallerThanRight;
                    if (timeIn) {
                        break;
                    }
                } catch (NumberFormatException e) {
                    timeIn = false;
                    break;
                }
            } else {
                timeIn = false;
                break;
            }
        }
        return timeIn;
    }

    public void analyze(ZooKeeper zooKeeper, List<AbnormalJob> oldAbnormalJobs,
                        String jobName, String jobDegree, NameSpaceConfiguration config) {
        AbnormalJob unnormalJob = new AbnormalJob(jobName, config.getNameSpace(), config.getDegree());
        unnormalJob.setJobDegree(jobDegree);
        checkOutdatedNoRunningJob(oldAbnormalJobs, zooKeeper, unnormalJob);
    }

    private synchronized boolean contains(AbnormalJob abnormalJob) {
        return outdatedNoRunningJobs.contains(abnormalJob);
    }

    private synchronized void addAbnormalJob(AbnormalJob abnormalJob) {
        outdatedNoRunningJobs.add(abnormalJob);
    }

    private synchronized void addNeedReportAlarmJob(AbnormalJob abnormalJob) {
        needReportAlarmJobs.add(abnormalJob);
    }

    private void checkOutdatedNoRunningJob(List<AbnormalJob> oldAbnormalJobs,
                                           ZooKeeper zooKeeper, AbnormalJob abnormalJob) {
        try {
            if (!isCronJob(zooKeeper, abnormalJob.getJobName())) {
                return;
            }
            if (!isEnabledPath(zooKeeper, abnormalJob)) {
                return;
            }
            doCheckAndHandleOutdatedNoRunningJob(oldAbnormalJobs, zooKeeper, abnormalJob);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void doCheckAndHandleOutdatedNoRunningJob(List<AbnormalJob> oldAbnormalJobs,
                                                      ZooKeeper zooKeeper, AbnormalJob abnormalJob) {
        String jobName = abnormalJob.getJobName();
        String enabledPath = TaskNodePath.getConfigNodePath(abnormalJob.getJobName(), "enabled");
        List<String> items = zooKeeper.getChildren(TaskNodePath.getExecutionNodePath(abnormalJob.getJobName()));
        if (items != null && !items.isEmpty()) { // 有分片
            int shardingTotalCount = Integer.parseInt(zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, "shardingTotalCount")));
            for (String item : items) {
                int each = Integer.parseInt(item);
                if (each < shardingTotalCount) { // 过滤历史遗留分片
                    doCheckAndHandleOutdatedNoRunningJobByShardingItem(oldAbnormalJobs, zooKeeper, abnormalJob, enabledPath, item);
                }
            }
        } else { // 无分片。还没有开始第一次的作业执行。
            abnormalJob.setCause(AbnormalJob.Cause.NO_SHARDS.name());
            long nextFireTimeAfterThis = zooKeeper.getMtime(enabledPath);
            Long nextFireTime = getNextFireTimeAfterSpecifiedTimeExcludePausePeriod(nextFireTimeAfterThis, jobName, zooKeeper);
            // 下次触发时间是否小于当前时间+延时, 是则为过时未跑有异常
            if (nextFireTime != null && nextFireTime + DashboardConstants.ALLOW_DELAY_MILLIONSECONDS < System.currentTimeMillis()) {
                handleOutdatedNoRunningJob(oldAbnormalJobs, zooKeeper, abnormalJob, nextFireTime);
            }
        }
    }

    /**
     * 检查和处理问题作业
     */
    private void doCheckAndHandleOutdatedNoRunningJobByShardingItem(List<AbnormalJob> oldAbnormalJobs,
                                                                    ZooKeeper zooKeeper, AbnormalJob abnormalJob, String enabledPath,
                                                                    String item) {
        if (contains(abnormalJob)) {
            return;
        }
        String jobName = abnormalJob.getJobName();
        int cversion = getCversion(zooKeeper, TaskNodePath.getExecutionItemNodePath(jobName, item));
        long nextFireTime = checkShardingItemState(zooKeeper, abnormalJob, enabledPath, item);
        if (nextFireTime != -1 && doubleCheckShardingState(abnormalJob, item, cversion)
                && !mayBlockWaitingRunningItemEnd(zooKeeper, abnormalJob, nextFireTime, item)) {
            if (abnormalJob.getCause() == null) {
                abnormalJob.setCause(AbnormalJob.Cause.NOT_RUN.name());
            }
            handleOutdatedNoRunningJob(oldAbnormalJobs, zooKeeper, abnormalJob, nextFireTime);
        }
    }

    /**
     * 判断分片状态
     * <p>
     * 逻辑： 1、注意：针对stock-update域的不上报节点信息但又有分片残留的情况，分片节点下只有两个子节点，返回正常 2、有running节点，返回正常
     * 3.1、有completed节点，但马上就取不到Mtime，节点有变动说明正常 3.2、根据Mtime计算下次触发时间，比较下次触发时间是否小于当前时间+延时, 是则为过时未跑有异常
     * 4、既没有running又没completed视为异常
     *
     * @return -1：状态正常，非-1：状态异常
     */
    private long checkShardingItemState(ZooKeeper zooKeeper,
                                        AbnormalJob abnormalJob, String enabledPath, String shardingItemStr) {
        List<String> itemChildren = zooKeeper
                .getChildren(TaskNodePath.getExecutionItemNodePath(abnormalJob.getJobName(), shardingItemStr));

        // 注意：针对stock-update域的不上报节点信息但又有分片残留的情况，分片节点下只有两个子节点，返回正常
        if (itemChildren.size() == 2) {
            return -1;
        }
        // 有running节点，返回正常
        if (itemChildren.contains("running")) {
            return -1;
        }

        // 有completed节点：尝试取分片节点的Mtime时间
        // 1、能取到则根据Mtime计算下次触发时间，比较下次触发时间是否小于当前时间+延时, 是则为过时未跑有异常
        // 2、取不到（为0）说明completed节点刚好被删除了，节点有变动说明正常（上一秒还在，下一秒不在了）
        if (itemChildren.contains("completed")) {
            return checkShardingItemStateWhenIsCompleted(zooKeeper, abnormalJob, enabledPath, shardingItemStr);
        } else { // 既没有running又没completed视为异常
            return checkShardingItemStateWhenNotCompleted(zooKeeper, abnormalJob, enabledPath,
                    shardingItemStr);
        }
    }

    private long checkShardingItemStateWhenIsCompleted(ZooKeeper zooKeeper,
                                                       AbnormalJob abnormalJob, String enabledPath, String shardingItemStr) {
        long currentTime = System.currentTimeMillis();
        String completedPath = TaskNodePath.getExecutionNodePath(abnormalJob.getJobName(), shardingItemStr, "completed");
        long completedMtime = zooKeeper.getMtime(completedPath);

        if (completedMtime > 0) {
            // 对比minCompletedMtime与enabled mtime, 取最大值
            long nextFireTimeAfterThis = zooKeeper.getMtime(enabledPath);
            if (nextFireTimeAfterThis < completedMtime) {
                nextFireTimeAfterThis = completedMtime;
            }

            Long nextFireTimeExcludePausePeriod = getNextFireTimeAfterSpecifiedTimeExcludePausePeriod(
                    nextFireTimeAfterThis, abnormalJob.getJobName(), zooKeeper);
            // 下次触发时间是否小于当前时间+延时, 是则为过时未跑有异常
            if (nextFireTimeExcludePausePeriod != null
                    && nextFireTimeExcludePausePeriod + DashboardConstants.ALLOW_DELAY_MILLIONSECONDS < currentTime) {
                // 为了避免误报情况，加上一个delta，然后再计算
                if (!doubleCheckShardingStateAfterAddingDeltaInterval(zooKeeper, abnormalJob,
                        nextFireTimeAfterThis, nextFireTimeExcludePausePeriod, currentTime)) {
                    log.debug("still has problem after adding delta interval");
                    return nextFireTimeExcludePausePeriod;
                } else {
                    return -1;
                }
            }
        }
        return -1;
    }

    private long checkShardingItemStateWhenNotCompleted(ZooKeeper zooKeeper,
                                                        AbnormalJob abnormalJob, String enabledPath, String shardingItemStr) {
        if (abnormalJob.getNextFireTimeAfterEnabledMtimeOrLastCompleteTime() == 0) {
            long nextFireTimeAfterThis = zooKeeper.getMtime(enabledPath);
            long lastCompleteTime = getLastCompleteTime(zooKeeper, abnormalJob.getJobName(), shardingItemStr);
            if (nextFireTimeAfterThis < lastCompleteTime) {
                nextFireTimeAfterThis = lastCompleteTime;
            }

            abnormalJob.setNextFireTimeAfterEnabledMtimeOrLastCompleteTime(
                    getNextFireTimeAfterSpecifiedTimeExcludePausePeriod(nextFireTimeAfterThis, abnormalJob.getJobName(),
                            zooKeeper));
        }
        Long nextFireTime = abnormalJob.getNextFireTimeAfterEnabledMtimeOrLastCompleteTime();
        // 下次触发时间是否小于当前时间+延时, 是则为过时未跑有异常
        if (nextFireTime != null && nextFireTime + DashboardConstants.ALLOW_DELAY_MILLIONSECONDS < System
                .currentTimeMillis()) {
            return nextFireTime;
        }
        return -1;
    }

    /**
     * 符合连续两次告警返回true，否则返回false ALLOW_DELAY_MILLIONSECONDS * 1.5 钝化触发检查的时间窗口精度 告警触发条件： 1、上次告警+本次检查窗口告警（连续2次）
     * 2、上次告警CVersion与本次一致（说明当前本次检查窗口时间内没有子节点变更）
     */
    private boolean doubleCheckShardingState(AbnormalJob abnormalJob, String shardingItemStr, int zkNodeCVersion) {
        String key = abnormalJob.getNamespace() + "_" + abnormalJob.getJobName() + "_" + shardingItemStr;
        long nowTime = System.currentTimeMillis();

        if (abnormalShardingStateCache.containsKey(key)) {
            AbnormalShardingState abnormalShardingState = abnormalShardingStateCache.get(key);
            if (abnormalShardingState != null
                    && abnormalShardingState.getAlertTime() + DashboardConstants.ALLOW_DELAY_MILLIONSECONDS * 1.5
                    > nowTime && abnormalShardingState.getZkNodeCVersion() == zkNodeCVersion) {
                abnormalShardingStateCache.put(key, new AbnormalShardingState(nowTime, zkNodeCVersion));// 更新告警
                return true;
            } else {
                abnormalShardingStateCache.put(key, new AbnormalShardingState(nowTime, zkNodeCVersion));// 更新无效（过时）告警
                return false;
            }
        } else {
            abnormalShardingStateCache.put(key, new AbnormalShardingState(nowTime, zkNodeCVersion));// 新增告警信息
            return false;
        }
    }

    /**
     * 判断是否可能在等待其他处于running状态的item运行结束
     * <p>
     * 假如当前job有其它item在running，在以下情形下会导致当前item无法准点运行
     * 1.当console下发重新分片时，会导致已经跑完的item等待running的item
     * 2.executor会优先运行failover的item，可能会导致自己的item无法准点运行
     * 3.需要被failover的item可能由于所有的executor都在running自己的分片，导致不能被及时failover
     * 可能不止以上的情况
     * <p>
     * 处理方式为:
     * 在当前job有其它running的item时
     * 1.考虑failover节点mtime
     * 2.进一步增加告警延迟时间
     *
     * @return 返回false会触发告警
     */
    private boolean mayBlockWaitingRunningItemEnd(ZooKeeper zooKeeper,
                                                  AbnormalJob abnormalJob, long nextFireTime, String shardingItemStr) {
        List<String> executionItems = zooKeeper
                .getChildren(TaskNodePath.getExecutionNodePath(abnormalJob.getJobName()));

        boolean hasRunningItem = false;
        if (!CollectionUtils.isEmpty(executionItems)) {
            for (String item : executionItems) {
                String runningNodePath = TaskNodePath.getRunningNodePath(abnormalJob.getJobName(), item);
                if (zooKeeper.checkExists(runningNodePath)) {
                    hasRunningItem = true;
                    break;
                }
            }
        }

        if (!hasRunningItem) {
            return false;
        }

        //假如failover节点mtime大于nextFireTime，则更新nextFireTime
        long currentTime = System.currentTimeMillis();
        long failoverMtime;
        String failoverNodePath = TaskNodePath.getFailoverNodePath(abnormalJob.getJobName(), shardingItemStr);
        if ((failoverMtime = zooKeeper.getMtime(failoverNodePath)) <= 0) {
            String leaderFailoverItemPath = TaskNodePath
                    .getLeaderFailoverItemPath(abnormalJob.getJobName(), shardingItemStr);
            failoverMtime = zooKeeper.getMtime(leaderFailoverItemPath);
        }

        long nextFireTimeTmp = failoverMtime > nextFireTime ? failoverMtime : nextFireTime;
        return nextFireTimeTmp + DashboardConstants.NOT_RUNNING_WARN_DELAY_MS_WHEN_JOB_RUNNING > currentTime;
    }


    private int getCversion(ZooKeeper zooKeeper, String path) {
        int cversion = 0;
        Stat stat = zooKeeper.getStat(path);
        if (stat != null) {
            cversion = stat.getCversion();
        }
        return cversion;
    }


    private void handleOutdatedNoRunningJob(List<AbnormalJob> oldAbnormalJobs,
                                            ZooKeeper zooKeeper, AbnormalJob abnormalJob, Long nextFireTime) {

        // 补充异常信息
        fillAbnormalJobInfo(zooKeeper, oldAbnormalJobs, abnormalJob, abnormalJob.getCause(),
                nextFireTime);
        // 如果有必要，告警
        reportAlarmAbnormalJobIfNecessary(zooKeeper, abnormalJob);
        addAbnormalJob(abnormalJob);
    }

    private void fillAbnormalJobInfo(ZooKeeper zooKeeper,
                                     List<AbnormalJob> oldAbnormalJobs, AbnormalJob abnormalJob, String cause,
                                     long nextFireTime) {
        if (executorNotReady(zooKeeper, abnormalJob)) {
            cause = AbnormalJob.Cause.EXECUTORS_NOT_READY.name();
        }
        abnormalJob.setCause(cause);

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        sdf.setTimeZone(TimeZone.getDefault());
        abnormalJob.setNextFireTimeWithTimeZoneFormat(sdf.format(nextFireTime));
        abnormalJob.setNextFireTime(nextFireTime);
        // 对比数据库中的数据，更新相关字段
        AbnormalJob oldAbnormalJob = DashboardServiceHelper.findEqualAbnormalJob(abnormalJob, oldAbnormalJobs);
        if (oldAbnormalJob != null) {
            abnormalJob.setRead(oldAbnormalJob.isRead());
            if (oldAbnormalJob.getUuid() != null) {
                abnormalJob.setUuid(oldAbnormalJob.getUuid());
            } else {
                abnormalJob.setUuid(UUID.randomUUID().toString());
            }
            abnormalJob.setHasRerun(oldAbnormalJob.isHasRerun());
        } else {
            abnormalJob.setUuid(UUID.randomUUID().toString());
        }
    }

    private boolean executorNotReady(ZooKeeper zooKeeper, AbnormalJob abnormalJob) {
        String jobName = abnormalJob.getJobName();
        String serverNodePath = TaskNodePath.getExecutorsNodePath(jobName);
        if (zooKeeper.checkExists(serverNodePath)) {
            List<String> servers = zooKeeper.getChildren(serverNodePath);
            if (servers != null && !servers.isEmpty()) {
                for (String server : servers) {
                    if (zooKeeper.checkExists(TaskNodePath.getExecutorStatus(jobName, server))) {
                        return false;
                    }
                }
            }
        }
        return true;
    }

    private void reportAlarmAbnormalJobIfNecessary(ZooKeeper zooKeeper,
                                                   AbnormalJob abnormalJob) {
        boolean rerun = getRerun(abnormalJob.getJobName(), zooKeeper);
        if (rerun) {
            if (abnormalJob.isHasRerun()) {
                reportAlarmIfNotRead(abnormalJob);
            } else {
                String namespace = abnormalJob.getNamespace();
                String jobName = abnormalJob.getJobName();
                try {
                    log.warn("found abnormal job:{}, will runAtOnce", abnormalJob);
                    taskService.runAtOnce(namespace, jobName);
                } catch (Throwable t) {
                    log.warn(String.format("rerun job error, namespace:%s, jobName:%s", namespace, jobName), t);
                    reportAlarmIfNotRead(abnormalJob);
                } finally {
                    abnormalJob.setHasRerun(true);
                }
            }
        } else {
            reportAlarmIfNotRead(abnormalJob);
        }
    }

    private boolean getRerun(String jobName, ZooKeeper zooKeeper) {
        String rerun = zooKeeper.getData(TaskNodePath.getConfigNodePath(jobName, TaskServiceImpl.CONFIG_ITEM_RERUN));
        return Boolean.valueOf(rerun);
    }

    private void reportAlarmIfNotRead(AbnormalJob abnormalJob) {
        if (!abnormalJob.isRead()) {
            addNeedReportAlarmJob(abnormalJob);
            // 兼容低版本，仍然可以使用单个作业告警
            String namespace = abnormalJob.getNamespace();
            String jobName = abnormalJob.getJobName();
            try {
                reportAlarmService.dashboardAbnormalJob(namespace, jobName, null,
                        abnormalJob.getNextFireTime());
            } catch (Throwable t) {
                log.error(
                        String.format("report alarm abnormal job error, namespace:%s, jobName:%s", namespace, jobName),
                        t);
            }
        }
    }


    private Long getNextFireTimeAfterSpecifiedTimeExcludePausePeriod(long nextFireTimeAfterThis, String jobName,
                                                                     ZooKeeper zooKeeper) {
        String cronPath = TaskNodePath.getConfigNodePath(jobName, "cron");
        String cronVal = zooKeeper.getData(cronPath);
        CronExpression cronExpression = null;
        try {
            cronExpression = new CronExpression(cronVal);
        } catch (ParseException e) {
            log.error(e.getMessage(), e);
            return null;
        }

        TimeZone timeZone = TimeZone.getDefault();
        cronExpression.setTimeZone(timeZone);

        Date nextFireTime = cronExpression.getTimeAfter(new Date(nextFireTimeAfterThis));
        String pausePeriodDatePath = TaskNodePath.getConfigNodePath(jobName, "pausePeriodDate");
        String pausePeriodDate = zooKeeper.getData(pausePeriodDatePath);
        String pausePeriodTimePath = TaskNodePath.getConfigNodePath(jobName, "pausePeriodTime");
        String pausePeriodTime = zooKeeper.getData(pausePeriodTimePath);

        while (nextFireTime != null && isInPausePeriod(nextFireTime, pausePeriodDate, pausePeriodTime, timeZone)) {
            nextFireTime = cronExpression.getTimeAfter(nextFireTime);
        }
        if (null == nextFireTime) {
            return null;
        }
        return nextFireTime.getTime();
    }

    /**
     * 为了避免executor时钟比Console快的现象，加上一个修正值，然后计算新的nextFireTime + ALLOW_DELAY_MILLIONSECONDS 依然早于当前时间。
     *
     * @return false: 依然有异常；true: 修正后没有异常。
     */
    private boolean doubleCheckShardingStateAfterAddingDeltaInterval(
            ZooKeeper zooKeeper, AbnormalJob abnormalJob,
            long nextFireTimeAfterThis, Long nextFireTimeExcludePausePeriod, long currentTime) {
        Long nextFireTimeExcludePausePeriodWithDelta = getNextFireTimeAfterSpecifiedTimeExcludePausePeriod(
                nextFireTimeAfterThis + DashboardConstants.INTERVAL_DELTA_IN_SECOND, abnormalJob.getJobName(),
                zooKeeper);

        if (nextFireTimeExcludePausePeriod.equals(nextFireTimeExcludePausePeriodWithDelta)
                || nextFireTimeExcludePausePeriodWithDelta + DashboardConstants.ALLOW_DELAY_MILLIONSECONDS
                < currentTime) {
            log.debug("still not work after adding delta interval");
            return false;
        }

        return true;
    }


    /**
     * 批量告警
     */
    public void reportAlarmOutdatedNoRunningJobs() {
        Map<String, List<Map<String, String>>> abnormalJobMap = new HashMap<>();
        for (AbnormalJob abnormalJob : needReportAlarmJobs) {
            String namespace = abnormalJob.getNamespace();
            if (!abnormalJobMap.containsKey(namespace)) {
                abnormalJobMap.put(namespace, new ArrayList<Map<String, String>>());
            }
            Map<String, String> customMap = new HashMap<>();
            customMap.put("job", abnormalJob.getJobName());

            customMap.put("shouldFiredTime", String.valueOf(abnormalJob.getNextFireTime()));
            abnormalJobMap.get(namespace).add(customMap);
        }
        for (String namespace : abnormalJobMap.keySet()) {
            raiseAlarmPerNamespace(abnormalJobMap.get(namespace), namespace);
        }
    }

    private void raiseAlarmPerNamespace(List<Map<String, String>> jobs, String namespace) {
        try {
            reportAlarmService.dashboardAbnormalBatchJobs(namespace, jobs);
        } catch (RuntimeException e) {
            log.error(String.format("batch report alarm abnormal job error, namespace:%s, jobs:%s", namespace,
                    getJobNamesString(jobs)), e);
        }
    }

    private String getJobNamesString(List<Map<String, String>> jobs) {
        StringBuilder jobNames = new StringBuilder();
        for (int i = 0, size = jobs.size(); i < size; i++) {
            Map<String, String> job = jobs.get(i);
            jobNames.append(job.get("job"));
            if (i < size - 1) {
                jobNames.append(',');
            }
        }
        return jobNames.toString();
    }

    public List<AbnormalJob> getOutdatedNoRunningJobs() {
        return new ArrayList<>(outdatedNoRunningJobs);
    }

    public void setAbnormalShardingStateCache(Map<String, AbnormalShardingState> abnormalShardingStateCache) {
        this.abnormalShardingStateCache = abnormalShardingStateCache;
    }

    public void setReportAlarmService(ReportAlarmService reportAlarmService) {
        this.reportAlarmService = reportAlarmService;
    }

    public void setTaskService(TaskService taskService) {
        this.taskService = taskService;
    }
}
